import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * Hello world!
 */
public class DurableTopicConsumer2 {
 
    public static void main(String[] args) throws Exception {
        thread(new HelloWorldConsumer(), false);
    }
 
    public static void thread(Runnable runnable, boolean daemon) {
        Thread brokerThread = new Thread(runnable);
        brokerThread.setDaemon(daemon);
        brokerThread.start();
    }
 
    public static class HelloWorldProducer implements Runnable {
        public void run() {
            try {


            }
            catch (Exception e) {
                System.out.println("Caught: " + e);
                e.printStackTrace();
            }
        }
    }
 
    public static class HelloWorldConsumer implements Runnable, ExceptionListener {
        public void run() {
            try {

                ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.17:61616?jms.prefetchPolicy.all=20");

                // Create a Connection
                Connection connection = connectionFactory.createConnection();
                //持久订阅要设置clientId
                connection.setClientID("durableConnection2");
                connection.start();

                connection.setExceptionListener(this);

                // Create a Session
                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);


                // Create the destination (Topic or Queue)
                Topic topic = session.createTopic("TEST.DURABLETOPIC");

                // 需要创建持久化subscriber
                MessageConsumer consumer = session.createDurableSubscriber(topic, "MQ_sub");
                //使用监听器 来接收消息 ，要求 connect 不能断掉，一旦断掉 ，则 无法接受到消息
                consumer.setMessageListener(new MessageListener() {
                    @Override
                    public void onMessage(Message message) {
                        if (message instanceof TextMessage) {
                            TextMessage textMessage = (TextMessage) message;

                            String text = null;
                            try {
                                text = textMessage.getText();
                                System.out.println("Received: " + text);
                                Thread.currentThread().sleep(100);
                                textMessage.acknowledge();
                            } catch (JMSException e) {
                                e.printStackTrace();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }

                        }

                    }
                });

            } catch (Exception e) {

                System.out.println("Caught: " + e);
                e.printStackTrace();
            }
        }
 
        public synchronized void onException(JMSException ex) {
            System.out.println("JMS Exception occured.  Shutting down client.");
        }
    }
}